package org.xxd.kafka.clients.producer;

import org.xxd.kafka.clients.common.CommonClientConfigs;
import org.xxd.kafka.clients.common.util.Utils;

import java.net.InetSocketAddress;
import java.util.Arrays;
import java.util.List;
import java.util.Properties;

/**
 * @author: XiaoDong.Xie
 * @create: 2020-09-27 10:58
 * @description:
 */
public class ProducerConfig {
    public static final String TIMEOUT_CONFIG = "timeout.ms";
    public static final String CLIENT_ID_CONFIG = CommonClientConfigs.CLIENT_ID_CONFIG;
    public static final String PARTITIONER_CLASS_CONFIG = "partitioner.class";
    public static final String RETRY_BACKOFF_MS_CONFIG = CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG;
    public static final String METADATA_MAX_AGE_CONFIG = CommonClientConfigs.METADATA_MAX_AGE_CONFIG;
    public static final String MAX_REQUEST_SIZE_CONFIG = "max.request.size";
    public static final String BUFFER_MEMORY_CONFIG = "buffer.memory";
    public static final String COMPRESSION_TYPE_CONFIG = "compression.type";
    public static final String MAX_BLOCK_MS_CONFIG = "max.block.ms";
    public static final String REQUEST_TIMEOUT_MS_CONFIG = CommonClientConfigs.REQUEST_TIMEOUT_MS_CONFIG;
    public static final String BATCH_SIZE_CONFIG = "batch.size";
    public static final String LINGER_MS_CONFIG = "linger.ms";
    public static final String BOOTSTRAP_SERVERS_CONFIG = CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG;
    /** <code>connections.max.idle.ms</code> */
    public static final String CONNECTIONS_MAX_IDLE_MS_CONFIG = CommonClientConfigs.CONNECTIONS_MAX_IDLE_MS_CONFIG;
    public static final String MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION = "max.in.flight.requests.per.connection";
    public static final String RECONNECT_BACKOFF_MS_CONFIG = CommonClientConfigs.RECONNECT_BACKOFF_MS_CONFIG;
    public static final String SEND_BUFFER_CONFIG = CommonClientConfigs.SEND_BUFFER_CONFIG;
    public static final String RECEIVE_BUFFER_CONFIG = CommonClientConfigs.RECEIVE_BUFFER_CONFIG;
    public static final String ACKS_CONFIG = "acks";
    public static final String RETRIES_CONFIG = "retries";

    private Properties properties;

    public ProducerConfig(Properties properties) {
        this.properties = properties;
    }

    public String getString(String key, String defaultValue) {
        return properties.getProperty(key) == null
                ? defaultValue
                : properties.getProperty(TIMEOUT_CONFIG);
    }

    /**
     * 获取配置类
     *
     * @param key
     * @param t
     * @param <T>
     * @return
     */
    public <T> T getConfiguredInstance(String key, Class<T> t) {
        Class<?> c = getClass(key);
        if (c == null) {
            try {
                return t.newInstance();
            } catch (InstantiationException e) {
                e.printStackTrace();
            } catch (IllegalAccessException e) {
                e.printStackTrace();
            }
        }
        return (T) Utils.newInstance(c);
    }

    public Class<?> getClass(String key) {
        String className = properties.getProperty(key);
        if (className == null) {
            return null;
        }
        try {
            return Class.forName(className);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
        }
        return null;
    }

    public long getLong(String key, long defaultValue) {
        return properties.getProperty(key) == null
                ? defaultValue
                : Long.parseLong(properties.getProperty(key));
    }

    public int getInt(String key, int defaultValue) {
        return properties.getProperty(key) == null
                ? defaultValue
                : Integer.parseInt(properties.getProperty(key));
    }

    public List<String> getList(String key) {
        String values = properties.getProperty(key);
        String[] result = values.split(",");
        if (result != null) {
            return Arrays.asList(result);
        }
        return null;
    }
}
